"""
AkShare 适配器

提供免费的中国股票市场数据，包括：
- 实时行情数据
- 历史K线数据（支持分钟级）
- 指数数据
- 财务数据

特点：
- 完全免费，无需API Token
- 支持分钟级K线（1m, 5m, 15m, 30m, 60m）
- 数据来源可靠
- 实时更新

作者：RedFire Team
创建时间：2025-10-06
"""

import asyncio
import logging
from datetime import datetime, timedelta
from decimal import Decimal
from typing import List, Optional, Any
import pandas as pd

try:
    import akshare as ak
except ImportError:
    ak = None

from ...core.base import HistoricalDataAdapter
from ...core.exceptions import (
    AdapterConnectionError,
    AdapterDataError,
    AdapterConfigError,
    AdapterTimeoutError,
)
from ...core.models import Bar, Tick, DataFrequency


class AkShareAdapterConfig:
    """AkShare适配器配置"""
    
    def __init__(
        self,
        timeout: int = 30,
        retry_count: int = 3,
        retry_delay: float = 1.0,
        use_cache: bool = True,
    ):
        """
        Args:
            timeout: 请求超时时间（秒）
            retry_count: 重试次数
            retry_delay: 重试延迟（秒）
            use_cache: 是否使用缓存
        """
        self.timeout = timeout
        self.retry_count = retry_count
        self.retry_delay = retry_delay
        self.use_cache = use_cache


class AkShareAdapter(HistoricalDataAdapter):
    """
    AkShare数据适配器
    
    支持功能：
    - 历史K线数据（日线、分钟线）
    - 实时行情数据
    - 指数数据
    
    使用示例：
        ```python
        config = AkShareAdapterConfig(timeout=30)
        adapter = AkShareAdapter(config)
        
        async with adapter:
            # 获取分钟K线
            bars = await adapter.query_bars(
                symbol="000001",  # 上证指数
                start_time=datetime(2025, 10, 6),
                end_time=datetime(2025, 10, 6),
                frequency="1m"
            )
        ```
    """
    
    def __init__(
        self,
        config: AkShareAdapterConfig,
        cache_manager: Optional[Any] = None,
        logger: Optional[logging.Logger] = None
    ):
        if ak is None:
            raise ImportError(
                "AkShare not installed. Install it with: pip install akshare"
            )
        
        self.config = config
        self.cache_manager = cache_manager
        self.logger = logger or logging.getLogger(__name__)
        
        self._connected = False
        
        # 频率映射：标准频率 -> AkShare参数
        self._frequency_mapping = {
            DataFrequency.MINUTE_1: "1",
            DataFrequency.MINUTE_5: "5",
            DataFrequency.MINUTE_15: "15",
            DataFrequency.MINUTE_30: "30",
            DataFrequency.MINUTE_60: "60",
            DataFrequency.DAILY: "daily",
            DataFrequency.WEEKLY: "weekly",
        }
    
    # ==================== 连接管理 ====================
    
    async def initialize(self) -> None:
        """初始化适配器（由AdapterManager调用）"""
        await self.connect()
    
    async def connect(self) -> None:
        """连接AkShare（无需实际连接，仅做检查）"""
        try:
            self.logger.info("Initializing AkShare adapter...")
            
            # 测试导入
            if ak is None:
                raise AdapterConnectionError("AkShare module not available")
            
            self._connected = True
            self.logger.info("AkShare adapter initialized successfully")
            
        except Exception as e:
            raise AdapterConnectionError(f"Failed to initialize AkShare: {e}")
    
    async def disconnect(self) -> None:
        """断开连接"""
        self._connected = False
        self.logger.info("AkShare adapter disconnected")
    
    async def cleanup(self) -> None:
        """清理资源（由AdapterManager调用）"""
        await self.disconnect()
    
    async def _test_connection(self) -> None:
        """测试连接（尝试获取少量数据）"""
        try:
            # 尝试获取上证指数最新数据
            await asyncio.to_thread(ak.stock_zh_index_spot_em)
            self.logger.info("AkShare connection test passed")
        except Exception as e:
            raise AdapterConnectionError(f"Connection test failed: {e}")
    
    # ==================== 数据查询方法 ====================
    
    async def query_bars(
        self,
        symbol: str,
        start_time: datetime,
        end_time: datetime,
        frequency: str = "1d",
        **kwargs
    ) -> List[Bar]:
        """
        查询K线数据
        
        Args:
            symbol: 标的代码，支持格式：
                   - "000001" 或 "000001.SH" (上证指数)
                   - "sh000001" (上证指数，AkShare格式)
            start_time: 开始时间
            end_time: 结束时间
            frequency: 数据频率 ('1m', '5m', '15m', '30m', '60m', '1d', '1w')
        
        Returns:
            List[Bar]: K线数据列表
        """
        if not self._connected:
            raise AdapterConnectionError("AkShare not connected")
        
        # 转换频率
        freq_enum = DataFrequency.from_string(frequency)
        if freq_enum not in self._frequency_mapping:
            raise NotImplementedError(f"AkShare不支持频率: {frequency}")
        
        akshare_freq = self._frequency_mapping[freq_enum]
        
        try:
            # 标准化symbol格式
            normalized_symbol = self._normalize_symbol(symbol)
            
            # 根据频率选择不同的获取方法
            if freq_enum == DataFrequency.DAILY:
                df = await self._get_daily_bars(normalized_symbol, start_time, end_time)
            elif freq_enum == DataFrequency.WEEKLY:
                df = await self._get_weekly_bars(normalized_symbol, start_time, end_time)
            elif freq_enum in [DataFrequency.MINUTE_1, DataFrequency.MINUTE_5, 
                              DataFrequency.MINUTE_15, DataFrequency.MINUTE_30, 
                              DataFrequency.MINUTE_60]:
                df = await self._get_minute_bars(normalized_symbol, akshare_freq, start_time, end_time)
            else:
                raise NotImplementedError(f"不支持的频率: {frequency}")
            
            if df is None or df.empty:
                self.logger.warning(f"未查询到数据: {symbol} {frequency}")
                return []
            
            # 转换为Bar对象
            bars = self._convert_to_bars(df, symbol, frequency)
            
            self.logger.info(f"查询到{len(bars)}条K线: {symbol} {frequency}")
            return bars
            
        except Exception as e:
            self.logger.error(f"查询K线数据失败: {e}", exc_info=True)
            raise AdapterDataError(f"查询K线数据失败: {e}")
    
    async def query_ticks(
        self,
        symbol: str,
        start_time: datetime,
        end_time: datetime,
        **kwargs
    ) -> List[Tick]:
        """查询Tick数据（AkShare不直接支持）"""
        raise NotImplementedError("AkShare暂不支持Tick数据查询")
    
    async def health_check(self) -> bool:
        """健康检查"""
        try:
            await self._test_connection()
            return True
        except Exception as e:
            self.logger.error(f"Health check failed: {e}")
            return False
    
    def get_capability(self) -> dict:
        """获取适配器能力"""
        return {
            "adapter_name": "akshare",
            "supports_realtime": True,
            "supports_historical": True,
            "supports_tick": False,
            "supported_frequencies": [
                "1m", "5m", "15m", "30m", "60m", "1d", "1w"
            ],
            "supported_markets": ["CN_A_SHARES", "CN_INDICES"],
            "max_bars_per_request": 10000,
            "rate_limit_per_minute": 60,
        }
    
    # ==================== 私有方法 ====================
    
    def _normalize_symbol(self, symbol: str) -> str:
        """
        标准化symbol格式
        
        转换规则：
        - "000001.SH" -> "sh000001"
        - "399001.SZ" -> "sz399001"
        - "000001" -> "sh000001" (默认上证)
        - "HSI" -> "HSI" (港股指数保持原样)
        """
        # 港股指数直接返回
        if symbol.upper() in ["HSI", "^HSI"]:
            return symbol.upper()
        
        # 移除可能的后缀
        if "." in symbol:
            code, exchange = symbol.split(".")
            if exchange.upper() == "SH":
                return f"sh{code}"
            elif exchange.upper() == "SZ":
                return f"sz{code}"
        
        # 如果没有后缀，根据代码判断
        if symbol.startswith("6"):
            return f"sh{symbol}"
        else:
            # 默认深圳或上证指数
            if symbol.startswith("000001"):
                return f"sh{symbol}"
            return f"sz{symbol}"
    
    async def _get_daily_bars(
        self, 
        symbol: str, 
        start_time: datetime, 
        end_time: datetime
    ) -> Optional[pd.DataFrame]:
        """获取日线数据"""
        try:
            # 判断是否为港股指数
            if symbol.upper() in ["HSI", "^HSI"]:
                # 港股指数日线 - 尝试多个API
                try:
                    # 优先使用sina API
                    df = await asyncio.to_thread(
                        ak.stock_hk_index_daily_sina,
                        symbol="HSI"
                    )
                    if df is not None and not df.empty:
                        self.logger.info("使用sina API成功获取HSI数据")
                        return df
                except Exception as sina_error:
                    self.logger.warning(f"sina API获取HSI失败: {sina_error}")
                
                try:
                    # 备选使用eastmoney API
                    df = await asyncio.to_thread(
                        ak.stock_hk_index_daily_em,
                        symbol="HSI"
                    )
                    if df is not None and not df.empty:
                        self.logger.info("使用eastmoney API成功获取HSI数据")
                        return df
                except Exception as em_error:
                    self.logger.warning(f"eastmoney API获取HSI失败: {em_error}")
                
                # 如果都失败了，返回None
                self.logger.error("所有HSI API都失败了")
                return None
                
            # 判断是否为A股指数
            elif symbol.startswith("sh000") or symbol.startswith("sz399"):
                # A股指数日线
                df = await asyncio.to_thread(
                    ak.stock_zh_index_daily,
                    symbol=symbol
                )
            else:
                # A股股票日线
                df = await asyncio.to_thread(
                    ak.stock_zh_a_hist,
                    symbol=symbol.replace("sh", "").replace("sz", ""),
                    period="daily",
                    start_date=start_time.strftime("%Y%m%d"),
                    end_date=end_time.strftime("%Y%m%d"),
                    adjust="qfq"  # 前复权
                )
            
            return df
            
        except Exception as e:
            self.logger.error(f"获取日线数据失败: {e}")
            return None
    
    async def _get_weekly_bars(
        self, 
        symbol: str, 
        start_time: datetime, 
        end_time: datetime
    ) -> Optional[pd.DataFrame]:
        """获取周线数据"""
        try:
            df = await asyncio.to_thread(
                ak.stock_zh_a_hist,
                symbol=symbol.replace("sh", "").replace("sz", ""),
                period="weekly",
                start_date=start_time.strftime("%Y%m%d"),
                end_date=end_time.strftime("%Y%m%d"),
                adjust="qfq"
            )
            return df
            
        except Exception as e:
            self.logger.error(f"获取周线数据失败: {e}")
            return None
    
    async def _get_minute_bars(
        self,
        symbol: str,
        period: str,
        start_time: datetime,
        end_time: datetime
    ) -> Optional[pd.DataFrame]:
        """
        获取分钟K线数据
        
        Args:
            symbol: 标的代码（如 "sh000001"）
            period: 周期（"1", "5", "15", "30", "60"）
            start_time: 开始时间
            end_time: 结束时间
            
        Note:
            AkShare的分钟线数据有时间限制，通常只返回最近几天的数据
            建议查询最近7天内的数据
        """
        try:
            # 提取纯代码
            code = symbol.replace("sh", "").replace("sz", "")
            
            # AkShare分钟线数据有时间限制，调整查询范围
            # 确保查询最近7天的数据以提高成功率
            now = datetime.now()
            query_end = min(end_time, now)
            query_start = max(start_time, now - timedelta(days=7))
            
            self.logger.info(f"查询分钟K线: symbol={code}, period={period}, "
                           f"start={query_start.strftime('%Y-%m-%d %H:%M:%S')}, "
                           f"end={query_end.strftime('%Y-%m-%d %H:%M:%S')}")
            
            # 判断是否为指数
            if symbol.startswith("sh000") or symbol.startswith("sz399"):
                # 使用index_zh_a_hist_min_em获取指数分钟线
                self.logger.info(f"使用指数分钟K线接口: symbol={code}")
                df = await asyncio.to_thread(
                    ak.index_zh_a_hist_min_em,
                    symbol=code,
                    period=period,
                    start_date=query_start.strftime("%Y-%m-%d %H:%M:%S"),
                    end_date=query_end.strftime("%Y-%m-%d %H:%M:%S")
                )
            else:
                # 使用stock_zh_a_hist_min_em获取股票分钟线
                self.logger.info(f"使用股票分钟K线接口: symbol={code}")
                df = await asyncio.to_thread(
                    ak.stock_zh_a_hist_min_em,
                    symbol=code,
                    period=period,
                    start_date=query_start.strftime("%Y-%m-%d %H:%M:%S"),
                    end_date=query_end.strftime("%Y-%m-%d %H:%M:%S"),
                    adjust=""  # 不复权
                )
            
            if df is not None and len(df) > 0:
                self.logger.info(f"成功获取 {len(df)} 条分钟K线数据")
            else:
                self.logger.warning(f"未获取到分钟K线数据，可能该时间范围内无交易数据")
            
            return df
            
        except Exception as e:
            self.logger.error(f"获取分钟K线失败: {e}")
            return None
    
    def _convert_to_bars(
        self, 
        df: pd.DataFrame, 
        symbol: str, 
        frequency: str
    ) -> List[Bar]:
        """
        将AkShare DataFrame转换为Bar对象列表
        
        AkShare日线数据列名：
        - 日期, 开盘, 收盘, 最高, 最低, 成交量, 成交额 (中文列名)
        - date, open, high, low, close, volume (英文列名 - HSI等港股指数)
        
        AkShare分钟线数据列名：
        - 时间, 开盘, 收盘, 最高, 最低, 成交量, 成交额
        """
        bars = []
        
        try:
            # 统一列名（支持中文和英文两种格式）
            column_mapping = {
                # 中文列名
                '日期': 'timestamp',
                '时间': 'timestamp',
                '开盘': 'open',
                '收盘': 'close',
                '最高': 'high',
                '最低': 'low',
                '成交量': 'volume',
                '成交额': 'amount',
                # 英文列名（港股指数等）
                'date': 'timestamp',
                'time': 'timestamp',
                # open, high, low, close, volume 保持不变，不需要映射
            }
            
            df = df.rename(columns=column_mapping)
            
            for _, row in df.iterrows():
                # 解析时间
                if isinstance(row['timestamp'], str):
                    # 尝试不同的时间格式
                    for fmt in ['%Y-%m-%d %H:%M:%S', '%Y-%m-%d', '%Y%m%d']:
                        try:
                            timestamp = datetime.strptime(row['timestamp'], fmt)
                            break
                        except ValueError:
                            continue
                    else:
                        # 如果都失败，尝试pandas解析
                        timestamp = pd.to_datetime(row['timestamp'])
                else:
                    timestamp = row['timestamp']
                
                bar = Bar(
                    symbol=symbol,
                    timestamp=timestamp,
                    open=Decimal(str(row['open'])),
                    high=Decimal(str(row['high'])),
                    low=Decimal(str(row['low'])),
                    close=Decimal(str(row['close'])),
                    volume=Decimal(str(row.get('volume', 0))),
                    amount=Decimal(str(row.get('amount', 0))) if row.get('amount') else None,
                    frequency=frequency
                )
                bars.append(bar)
            
            # 按时间排序
            bars.sort(key=lambda x: x.timestamp)
            
        except Exception as e:
            self.logger.error(f"数据转换失败: {e}", exc_info=True)
            raise AdapterDataError(f"数据转换失败: {e}")
        
        return bars

